package com.bw.func;

import com.alibaba.fastjson.JSON;
import com.alibaba.fastjson.JSONObject;
import com.bw.been.TableProcess;
import com.bw.common.GmallConfig;
import com.bw.util.ConfigUtil;
import com.bw.util.HBaseUtil;
import org.apache.flink.api.common.state.BroadcastState;
import org.apache.flink.api.common.state.MapStateDescriptor;
import org.apache.flink.api.common.state.ReadOnlyBroadcastState;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.streaming.api.functions.co.BroadcastProcessFunction;
import org.apache.flink.util.Collector;
import org.apache.flink.util.OutputTag;
import org.apache.hadoop.hbase.NamespaceDescriptor;
import org.apache.hadoop.hbase.TableName;
import org.apache.hadoop.hbase.client.*;
import org.apache.hadoop.hbase.util.Bytes;

import java.io.IOException;
import java.util.*;


public class TableProcessFunc extends BroadcastProcessFunction<JSONObject,String,JSONObject> {
    //定义侧输出流标记
    private final OutputTag<JSONObject> outputTag;

    private final MapStateDescriptor<String, TableProcess> mapStateDescriptor;

    public TableProcessFunc(OutputTag<JSONObject> outputTag,MapStateDescriptor<String,TableProcess> mapStateDescriptor){
        this.outputTag=outputTag;
        this.mapStateDescriptor=mapStateDescriptor;
    }

    //定义Hbase连接
    Connection connection=null;

    @Override
    public void open(Configuration parameters) throws Exception {
        connection = HBaseUtil.getConnection(ConfigUtil.getZookeeperQuorum(),ConfigUtil.getZookeeperClientPort());
    }

    @Override
    public void processElement(JSONObject jsonObject, BroadcastProcessFunction<JSONObject, String, JSONObject>.ReadOnlyContext readOnlyContext, Collector<JSONObject> collector) throws Exception {
        //取出广播流
        ReadOnlyBroadcastState<String, TableProcess> broadcastState = readOnlyContext.getBroadcastState(mapStateDescriptor);

        //拼接key,mysql其他数据
        String table = jsonObject.getString("table");
        String type = jsonObject.getString("type");
        String key = table + ":" + type;

        //获取每条数据对应的配置
        TableProcess tableProcess = broadcastState.get(key);


        //判断配置信息是否存在
        if (tableProcess != null) {

            //过滤字段
            filterColumn(jsonObject.getJSONObject("data"), tableProcess.getSinkColumns());

            //分流,Kafka数据写入主流,hbase
            String sinkType = tableProcess.getSinkType();

            //获取Kafka主题或者hbase表名
            String sinkTable = tableProcess.getSinkTable();
            jsonObject.put("sinkTable", sinkTable);

            if (TableProcess.SINK_TYPE_KAFKA.equals(sinkType)) {
                collector.collect(jsonObject);
            } else if (TableProcess.SINK_TYPE_HBASE.equals(sinkType)) {
                readOnlyContext.output(outputTag, jsonObject);
            }

        } else {
            System.out.println("配置信息中不存在Key:" + key);
        }
    }

    //处理广播流数据
    @Override
    public void processBroadcastElement(String s, BroadcastProcessFunction<JSONObject, String, JSONObject>.Context context, Collector<JSONObject> collector) throws Exception {
        //提取广播流数据,将string类型的数据转换为jsonObject
        JSONObject jsonObject = JSON.parseObject(s);

        //获取其中的data
        String data = jsonObject.getString("data");

        //将data转化成TableProcess格式
        TableProcess process = JSON.parseObject(data, TableProcess.class);

        if (process!=null){
            if (process.getSinkType().equals(TableProcess.SINK_TYPE_HBASE)){
                createHbaseTable(process);
            }

            //获取状态
            BroadcastState<String, TableProcess> broadcastState = context.getBroadcastState(mapStateDescriptor);

            //准备状态中的Key
            String sourceTable = process.getSourceTable();
            String operateType = process.getOperateType();
            String key = sourceTable + ":" + operateType;

            //将数据写入广播流
            broadcastState.put(key,process);
        }else {
            System.out.println("等待配置信息>>>>>");
        }
    }

    //根据指定的字段对数据进行过滤,校验字段，过滤掉多余的字段
    private void filterColumn(JSONObject data, String sinkColumns) {

        //获取data数据中的数据集
        Set<Map.Entry<String, Object>> entries = data.entrySet();

        //对需要保留的字段做切分并转换为List
        String[] columnsArr = sinkColumns.split(",");
        List<String> columnList = Arrays.asList(columnsArr);

        //遍历数据集,如果在需要保留的字段中不存在,则移除该字段
        entries.removeIf(next -> !columnList.contains(next.getKey()));
    }

    public void createHbaseTable(TableProcess tableProcess) throws IOException {
        String sinkTable=tableProcess.getSinkTable();
        String sinkColumns = tableProcess.getSinkColumns();

        Admin admin=null;

        try {
            admin=connection.getAdmin();
            //获取命名空间
            NamespaceDescriptor[] descriptors = admin.listNamespaceDescriptors();

            //判断 GmallConfig.HBASE_NAMESPASE 是否存在
            // noneMatch 存在返回 false
            //anyMatch 存在返回 true
            if (Arrays.stream(descriptors).noneMatch(a->a.getName().equals(GmallConfig.HBASE_SCHEMA))){
                //创建命名空间描述符
                NamespaceDescriptor descriptor = NamespaceDescriptor.create(GmallConfig.HBASE_SCHEMA)
                        .addConfiguration("gmall-2022-realtime", "保险业务实时数仓")
                        .build();
                //创建命名空间
                admin.createNamespace(descriptor);
            }

            TableName tableName = TableName.valueOf(GmallConfig.HBASE_SCHEMA + ":" + sinkTable);
            System.out.println(tableName+">>>>>>>");

            if(!admin.tableExists(tableName)){
                TableDescriptorBuilder builder = TableDescriptorBuilder.newBuilder(tableName);

                ColumnFamilyDescriptor familyDescriptor = ColumnFamilyDescriptorBuilder.newBuilder(Bytes.toBytes("data")).build();

                builder.setColumnFamily(familyDescriptor);

                TableDescriptor build = builder.build();

                admin.createTable(build);

                System.out.println("表"+sinkTable+"创建成功>>>>>>>>>");
            }else {
                System.out.println("表"+sinkTable+"已经存在>>>>>>>>>");
            }
        } catch (IOException e) {
            throw new RuntimeException(e);
        } finally {
            if (admin != null){
                try {
                    admin.close();
                } catch (IOException e) {
                    throw new RuntimeException(e);
                }
            }
        }
    }
}
